package AppBehavier;

import bean.MarketingUserBehavior;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * @author Spring_Hu
 * @date 2021/10/12 21:52
 */
public class AppBehavier01 {
    static StreamExecutionEnvironment env =null;
    //APP通过每个渠道每种行为的数量
    public static void main(String[] args) throws Exception {
        //造数据
        env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(new SourceFunction<MarketingUserBehavior>() {
            boolean canRun = true;
            Random random = new Random();
            List<String> channels = Arrays.asList("huawwei", "xiaomi", "apple", "baidu", "qq", "oppo", "vivo");
            List<String> behaviors = Arrays.asList("download", "install", "update", "uninstall");
            @Override
            public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
                while (canRun) {
                    MarketingUserBehavior marketingUserBehavior = new MarketingUserBehavior(
                            (long) random.nextInt(1000000),
                            behaviors.get(random.nextInt(behaviors.size())),
                            channels.get(random.nextInt(channels.size())),
                            System.currentTimeMillis());
                    ctx.collect(marketingUserBehavior);
                    Thread.sleep(1000);
                }

            }

            @Override
            public void cancel() {
                canRun=false;
            }
        }).map(value -> Tuple2.of(value.getBehavior()+"_"+value.getChannel(),1L))
                .returns(Types.TUPLE(Types.STRING,Types.LONG))
                .keyBy(value -> value.f0)
                .sum(1)
                .print();

        env.execute();
    }
}
